package com.atguigu.Flink.sql.query;

import com.atguigu.Flink.POJO.WindSpeedSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Flink08_OverWindowSQL {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WindSpeedSensor> ds = env.socketTextStream("hadoop102", 8888)
                .map(
                        line -> {
                            String[] split = line.split(",");
                            return new WindSpeedSensor(split[0], Integer.valueOf(split[1]), Long.valueOf(split[2]));
                        }
                );
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Schema schema = Schema.newBuilder()
                .column("id", "STRING")
                .column("speed" , "INT")
                .column("ts" , "BIGINT")
                .columnByExpression("pt" , "proctime()")
                .columnByExpression("et" , "TO_TIMESTAMP_LTZ(ts, 3) ")
                .watermark("et" ,  "et - INTERVAL '0' SECOND ")
                .build();
        Table table = tableEnv.fromDataStream(ds, schema);
        tableEnv.createTemporaryView("t1",table);

        //基于行
        //上两行，到当前行
        String sql1 =
                "select id,speed,sum(speed) over(partition by id order by pt " +
                        "rows between 2 preceding and current row) as " +
                        "sumspeed from t1";
        //上无边界，到当前行
        String sql2 =
                "select id,speed,sum(speed) over(partition by id order by pt " +
                        "rows between unbounded preceding and current row) as " +
                        "sumspeed from t1";

        //上2S，到当前行
        String sql3 =
                "select id,speed,pt,et,ts,sum(speed) over(partition by id order by pt " +
                        "RANGE BETWEEN INTERVAL '2' SECOND preceding and current row) as " +
                        "sumspeed from t1";

        //上无边界到当前行
        String sql4 =
                "select id,speed,pt,et,ts,sum(speed) over(partition by id order by pt " +
                        "RANGE BETWEEN unbounded preceding and current row) as " +
                        "sumspeed from t1";

        //多窗口聚合
        String sql5 =
                " select id, speed , ts , pt ," +
                        " sum(speed) over w as sumvc ," +
                        " max(speed) over w as maxvc, "+
                        " min(speed) over w as minvc  " +
                        //" 其他聚合() over(.....)"
                        " from t1 "  +
                        " window w as (partition by id order by pt RANGE BETWEEN UNBOUNDED PRECEDING  and CURRENT ROW)" ;

        tableEnv.sqlQuery(sql5).execute().print();


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
